#include "handler.h"
#include "request.h"
#include "response.h"
#include "message.h"
#include "channel.h"

Handler::Handler()
{
    m_running.store(true);
    m_waitThread = std::thread([=]() {
        waitThreadFunc();
    });
}

Handler::~Handler()
{
    m_running.store(false);
    m_emptyCond.notify_all();
    if(m_waitThread.joinable())
        m_waitThread.join();
}

void Handler::request(const std::shared_ptr<Request> &req)
{
    if(!req) {
        return;
    }
    if(!req->message() || !m_channel || !m_channel->sendMessage(req->message()))
    {
        auto res = std::make_shared<Response>();
        res->setState(Response::Fail);
        req->setResult(std::move(res));
        return;
    }
    if(m_channel)
    {
        std::lock_guard<std::mutex> lock(m_reqMutex);
        m_requests.emplace(req->message()->id(), req);
    }
    m_emptyCond.notify_one();
}

void Handler::response(std::shared_ptr<Response> &&res)
{
    if(!res || !res->message())
        return;
    std::lock_guard<std::mutex> lock(m_resMutex);
    m_responses.emplace(res->message()->id(), res);
}

void Handler::waitThreadFunc()
{
    while(m_running.load()) {
        {
            std::unique_lock<std::mutex> lock(m_reqMutex);
            m_emptyCond.wait(lock, [=]() { return !m_requests.empty() || !m_running.load(); });
            for(auto it = m_requests.begin(); it != m_requests.end(); ) {
                {
                    std::lock_guard<std::mutex> lock(m_resMutex);
                    auto resIt = m_responses.find(it->first);
                    if(resIt != m_responses.end()) {
                        Response::State state = resIt->second->state();
                        it->second->setResult(std::move(resIt->second));
                        m_responses.erase(resIt);
                        if(state != Response::InProgress)
                            m_requests.erase(it++);
                        continue;
                    }
                }
                auto now = std::chrono::steady_clock::now();
                if(it->second->isTimeout(now)) {
                    auto res = std::make_shared<Response>();
                    res->setState(Response::Timeout);
                    it->second->setResult(std::move(res));
                    m_requests.erase(it++);
                    continue;
                }
                it++;
            }
            {
                std::lock_guard<std::mutex> lock(m_resMutex);
                for(auto it = m_responses.begin(); it != m_responses.end(); ) {
                    if(m_requests.find(it->first) == m_requests.end()) {
                        m_responses.erase(it++);
                        continue;
                    }
                    it++;
                }
            }
            std::this_thread::sleep_for(std::chrono::milliseconds(1));
        }

    }
}

std::shared_ptr<Channel> Handler::channel() const
{
    return m_channel;
}

void Handler::setChannel(const std::shared_ptr<Channel> &channel)
{
    m_channel = channel;
}
